package com.gt.stream.req.util

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}

import java.util.Properties
import scala.collection.mutable.ListBuffer
import scala.util.Random

object Producer {

  def main(args: Array[String]): Unit = {

    //1. prepare data
    //example: timestamp area city userid adid
    //2. send msg

    val props = new Properties();
    props.put("bootstrap.servers", "localhost:9092");
    props.put("acks", "all");
    props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    val producer = new KafkaProducer[String, String](props)


    while (true){
      Thread.sleep(2000L)
      val msgList: ListBuffer[String] = createMsg()
      for(msg <- msgList){
        println(msg)
        producer.send(new ProducerRecord[String,String]("test",msg))
      }

    }


  }

  def createMsg(): ListBuffer[String] = {
    val msgList: ListBuffer[String] = ListBuffer()
    val areaList: ListBuffer[String] = ListBuffer("华北","华东","华中","华南")
    val cityList: ListBuffer[String] = ListBuffer("北京","上海","广州","深圳")

    for(i <- 0 to 10){
      val timestamp: Long = System.currentTimeMillis()
      val area: String = areaList.apply(new Random().nextInt(4))
      val city: String = cityList.apply(new Random().nextInt(4))
      val userid = new Random().nextInt(10)
      val adid = new Random().nextInt(10)

      val msg = s"$timestamp $area $city $userid $adid"
      msgList.append(msg)
    }
    msgList
  }

}
